package cn.steam.foo.service.impl;

import cn.steam.foo.message.Message;
import cn.steam.foo.service.KafkaService;
import cn.steam.foo.util.lock.KLock;
import com.alibaba.fastjson.JSON;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.ExecutionException;

/**
 * @author zhang jintao on 2022/05/18
 **/
@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    KafkaTemplate<String,String> kafkaTemplate;

    @Override
    @KLock(key = "")
    public void send(Message message) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(message.topic(), message.key(), JSON.toJSONString(message));
        SendResult<String, String> result = future.get();


    }

}
